/********************************************************************
*       (c) Copyright 2005, Hogeschool voor de Kunsten Utrecht
*                       Hilversum, the Netherlands
*********************************************************************
*
* File name     : msgqueue.cpp
* System name   : pthreads programming course
*
* Version       : $Revision: 1.1 $
*
*
* Description   : Use a message queue for communication between threads
*
*
* Author        : Marc_G
* E-mail        : marcg@dinkum.nl
*
********************************************************************/

/************
   $Log: msgqueue.cpp,v $
   Revision 1.1  2005/05/09 14:10:37  marcg
   new

Explanation:
When we have several threads running at different speeds and we want to send information in a specific order from one thread to another,
a message queue might be useful.

The idea is that the sending thread pushes messages into the queue and the receiving thread pops them off at its own pace.
This system works as long as the sending thread on average does not send more messages than the receiving thread can handle.
Because the queue works as a buffer, messages may be pushed and popped in bursts, in other words: there may be peaks in the traffic,
as long as the average sending speed over a period of time is lower than the capacity of the receiver.

The example shows an input thread that reads data from the console (cin) and pushes those onto the message queue.
The other thread, the receiver, checks the queue size and if that is not zero,
it pops a message off the queue and acts like it processes that message.
************/
#include <iostream>
#include <cstdlib>
#include <unistd.h>	// usleep
#include <fcntl.h>	// threads
#include <pthread.h>
#include <string>	// messages
#include <queue>	// the message queue

using namespace std;

pthread_mutex_t msgmutex = PTHREAD_MUTEX_INITIALIZER;
queue<string> msgq;

void *msgreceiver(void *arg)
{
    long qsize;
    string nextmsg;

    while(true){
        if(msgq.empty()){
            usleep(10000); // sleep 0.01 sec before trying again
            continue;
        }

        // we end up here because there was something in the msg queue
        pthread_mutex_lock(&msgmutex);
        qsize=msgq.size();
        if(qsize > 5)cout << "Queue size: " << qsize << endl;
        nextmsg=msgq.front(); // get next message in queue
        msgq.pop(); // remove it from the queue
        pthread_mutex_unlock(&msgmutex);

        cout << "Processing value :" << nextmsg << endl;
        usleep(2000000);
    }
    pthread_exit((void *)0);
} // msgreceiver()

void *msgtransmitter(void *arg)
{
    string nextmsg;

    while(true){
        cin >> nextmsg;
        pthread_mutex_lock(&msgmutex);
        msgq.push(nextmsg); // push message onto the queue
        pthread_mutex_unlock(&msgmutex);
    }
    pthread_exit((void *)0);
} // msgtransmitter()

int main(int argc, char **argv)
{
    pthread_t thr;

    // Create threads
    if(pthread_create(&thr,NULL,msgreceiver,NULL) || pthread_create(&thr,NULL,msgtransmitter,NULL))
    {
        cout << argv[0] << " cannot make thread\n";
        exit(1);
    }

    /*
   * At this point the main thread can perform its actions or end
   */
    cout << "** Main thread ends **\n";
    pthread_exit((void *)0);

} // main()

